package com.ls.fw.search.lucene.server;
import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.IndexWriterConfig.OpenMode;
import org.apache.lucene.index.IndexableField;
import org.apache.lucene.index.LogDocMergePolicy;
import org.apache.lucene.index.LogMergePolicy;
import org.apache.lucene.index.Term;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.SearcherFactory;
import org.apache.lucene.search.SearcherLifetimeManager;
import org.apache.lucene.search.SearcherManager;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.search.TopScoreDocCollector;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.store.NIOFSDirectory;
import org.apache.lucene.util.NamedThreadFactory;
import org.apache.lucene.util.Version;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ls.fw.data.exception.DataException;
import com.ls.fw.search.lucene.bean.SearchCondition;
import com.ls.fw.search.lucene.handler.SearchHandler;
import com.ls.fw.search.lucene.mapper.DocMapper;
import com.ls.fw.search.lucene.mapper.TopDocMapper;
import com.ls.fw.search.lucene.server.base.AbstractBaseServer;
import com.ls.fw.search.lucene.util.AnalyzerUtils;
import com.ls.fw.search.lucene.util.LuceneUtil;

/**
 * 
 * @author lisheng
 * @date 2015年1月2日 下午8:21:18
 * @version V1.0
 */
public class DefaultLuceneServer extends AbstractBaseServer  implements LuceneServer {

	public static final Version VERSION = Version.LATEST;

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private IndexWriter indexWriter = null;// 索引写入器

	private IndexReader indexReader = null;// 索引读取器

	private SearcherManager searcherManager;

	private SearcherLifetimeManager lifetimeMGR = new SearcherLifetimeManager();
	
	private Directory dir = null;

	private String address = "";// 索引文件路径

	private boolean opened = false;// 是否打开索引文件

	private boolean autoCommit = false;// 是否自动提交

	private boolean nearRealTime = true;
	

	private long time = System.currentTimeMillis();
	
	private byte[] lockWirter = new byte[1];
	private byte[] lockOpen = new byte[1];
	
	private ExecutorService service = null;

	{
		service = new ThreadPoolExecutor(5, 10, 10L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new NamedThreadFactory("IndexSearcher"));
	}
	
	
	private final SearcherFactory factory = new MySearcherFactory();
	 
	public class MySearcherFactory extends SearcherFactory{
		@Override
		public IndexSearcher newSearcher(IndexReader reader,
				IndexReader previousReader) throws IOException {
			IndexSearcher s = new IndexSearcher(reader, service);
			return s;
		}
	}
	
	public DefaultLuceneServer() {
		super();
	}

	public DefaultLuceneServer(String address) {
		super();
		this.address = address;
	}

	public DefaultLuceneServer(Directory dir) {
		super();
		this.dir = dir;
	}

	public DefaultLuceneServer(IndexWriter indexWriter, IndexReader indexReader) {
		super();
		this.indexWriter = indexWriter;
		this.dir = this.indexWriter.getDirectory();
		this.indexReader = indexReader;
	}

	
	@Override
	public IndexWriter getIndexWriter() {
		return this.indexWriter;
	}

	public void setIndexWriter(IndexWriter indexWriter) {
		this.indexWriter = indexWriter;
	}

	@Override
	public IndexReader getIndexReader() {
		return this.indexReader;
	}

	public void setIndexReader(IndexReader indexReader) {
		this.indexReader = indexReader;
	}

	@Override
	public void close() throws IOException {
		this.closeIndexReader();
		this.closeIndexWriter();
		if(searcherManager != null){
			searcherManager.close();
		}
		if(lifetimeMGR != null){
			lifetimeMGR.close();
		}
		if (dir != null) {
			this.dir.close();
		}
	}

	@Override
	public void closeIndexWriter() {
		if (this.indexWriter != null) {
			logger.info("关闭indexWriter....");
			try {
				this.indexWriter.close();
			} catch (IOException e) {
				throw new DataException(e.getMessage(), e);
			}
		}
		indexWriter = null;
	}

	@Override
	public void closeIndexReader() {
		if (this.indexReader != null) {
			logger.info("关闭IndexReader....");
			try {
				this.indexReader.close();
			} catch (IOException e) {
				throw new DataException(e.getMessage(), e);
			}
		}
		indexReader = null;
	}

	@Override
	public void rollback() {
		if (this.indexWriter != null) {
			logger.info("关闭indexWriter....");
			try {
				this.indexWriter.rollback();
			} catch (IOException e) {
				throw new DataException(e.getMessage(), e);
			}
		}
	}

	@Override
	public void commit() {
		this.commit(true);
	}
	
	public void commit(boolean softCommit){
		if(softCommit){
			Thread t = new Thread(new Runnable(){

				@Override
				public void run() {
					if (indexWriter != null) {
						try {
							logger.info("开始提交....");
							long start = System.currentTimeMillis();
							indexWriter.commit();
							logger.info("提交耗时:"+(System.currentTimeMillis()-start));
						} catch (IOException e) {
							throw new DataException(e.getMessage(), e);
						}
					}
				}
			});
			t.start();
		}else{
			if (this.indexWriter != null) {
				try {
					this.logger.info("开始提交....");
					long start = System.currentTimeMillis();
					this.indexWriter.commit();
					logger.info("提交耗时:"+(System.currentTimeMillis()-start));
				} catch (IOException e) {
					throw new DataException(e.getMessage(), e);
				}
			}
		}
	}

	/**
	 * @return the autoCommit
	 */
	@Override
	public boolean isAutoCommit() {
		return this.autoCommit;
	}

	/**
	 * @param autoCommit
	 *            the autoCommit to set
	 */
	public void setAutoCommit(boolean autoCommit) {
		this.autoCommit = autoCommit;
	}

	@Override
	public Directory getDir() {
		return dir;
	}

	@Override
	public Directory openDir(String path) {
		try {
			File file = new File(path);
			if(!file.exists()){
				file.mkdirs();
			}
			String name = System.getProperty("os.name");
			if(name != null && name.indexOf("Windows")>-1){
				this.dir = FSDirectory.open(Paths.get(path));
			}else{
				this.logger.info("使用NIO!");
				this.dir =  NIOFSDirectory.open(Paths.get(path));
			}
//			   NRTCachingDirectory cachedFSDir = new NRTCachingDirectory(fsDir, 5.0, 60.0);
//			this.dir = new RAMDirecstory(dir,IOContext.DEFAULT);

		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return dir;
	}

	@Override
	public String getAddress() {
		return this.address;
	}

	public LuceneServer setAddress(String address) {
		this.address = address;
		return this;
	}

	@Override
	public boolean isOpened() {
		return this.opened;
	}

	public void setOpened(boolean opened) {
		this.opened = opened;
	}

	public LuceneServer setDir(Directory dir) {
		this.dir = dir;
		return this;
	}

	@Override
	public boolean open() {
		return this.open(address);
	}

	@Override
	public boolean open(String address) {
		return this.open(address, AnalyzerUtils.getInstance(),
				OpenMode.CREATE_OR_APPEND);
	}
	


//	public void loadToMemery(){
//        /* 使用带参构造器创建 RAMDirectory 对象。下面代码参数为指向磁盘索引库的 FSDirectory 对象 */
//        try {
//			dir = new RAMDirectory(this.dir,new IOContext());
//		} catch (IOException e) {
//			throw new DataException(e.getMessage(), e);
//		}
//	}
	
//	public void refushToFs(){
//	  /*同步数据回磁盘 */
//	  Directory fsDir = FSDirectory.open(new File("./indexDir/"));
//	  IndexWriter  fsIndexWriter =  null;
//	  fsIndexWriter.addIndexesNoOptimize(RAMdir);
//	}
	
	public LuceneServer reOpenIndexReader() {
		this.closeIndexReader();
		synchronized (lockOpen) {
			if(indexReader==null){
				try {
					if(nearRealTime && this.indexWriter != null){
						this.indexReader = DirectoryReader.open(dir);
					}else{
						this.indexReader = DirectoryReader.open(this.indexWriter);
					}
				} catch (IOException e) {
					throw new DataException(e.getMessage(), e);
				}
			}
		}
		return this;
	}

	@Override
	public LuceneServer openIfChanged(){
		try {
			synchronized (lockOpen) {
				//reader.reopen(); // 读入新增加的增量索引内容，满足实时索引需求
	            IndexReader newReader = DirectoryReader.openIfChanged((DirectoryReader)this.indexReader
	            		, this.indexWriter, false);
	            if (newReader != null) {
	            	indexReader.close();
	            	indexReader = newReader;
	            }
			}
        } catch (Exception e) {
        	throw new DataException(e.getMessage(), e);
        }
		return this;
	}
	
	@Override
	public boolean open(String address, Analyzer analyzer, OpenMode openMode) {
		synchronized (lockOpen) {
			this.logger.info("打开服务"+address+"....");
			long start = System.currentTimeMillis();
			if (dir == null && StringUtils.isNotBlank(address)) {
				dir = this.openDir(address);
			}
			if (this.indexWriter == null && dir != null) {
				//如果目录下没有文件,则新建索引文件  
				File file = new File(this.address);
				if(file.isDirectory() && file.listFiles().length == 0){  
					// Create a new index in the directory, removing any
					// previously indexed documents:
					this.openIndexWriter(analyzer, OpenMode.CREATE);
					this.commit(false);
					this.closeIndexWriter();
				}  
				//如果已有索引文件则用原有的索引文件  OpenMode.CREATE_OR_APPEND
				// Add new documents to an existing index:
				this.openIndexWriter(analyzer, openMode);
			}
			if (this.indexReader == null && dir != null) {
				
				try {
					if(nearRealTime && this.indexWriter != null){
						this.indexReader = DirectoryReader.open(this.indexWriter);
					}else{
						this.indexReader = DirectoryReader.open(dir);
					}
				} catch (IOException e) {
					throw new DataException(e.getMessage(), e);
				}
			}
			try {
				if(searcherManager == null){
					if(nearRealTime && this.indexWriter != null){
						this.searcherManager = new SearcherManager(this.indexWriter,factory);
					}else{
						if(this.indexWriter != null){
							// SearcherManager needs to see empty commit:
							indexWriter.commit();
						}
						this.searcherManager = new SearcherManager(this.dir, factory);
					}
				}
			} catch (IOException e) {
				throw new DataException(e.getMessage(), e);
			}
			opened = true;
			logger.info("打开服务"+address+"耗时:"+(System.currentTimeMillis()-start));
		}
		return opened;
	}
	
	@Override
	public LuceneServer openIndexWriter(Analyzer analyzer, OpenMode openMode) {
		LogMergePolicy mergePolicy = new LogDocMergePolicy();  
        // 索引基本配置  
        // 设置segment添加文档(Document)时的合并频率  
        // 值较小,建立索引的速度就较慢  
        // 值较大,建立索引的速度就较快,>10适合批量建立索引  
        mergePolicy.setMergeFactor(30);  
        // 设置segment最大合并文档(Document)数  
        // 值较小有利于追加索引的速度  
        // 值较大,适合批量建立索引和更快的搜索  
        mergePolicy.setMaxMergeDocs(5000);  
		IndexWriterConfig iwc = new IndexWriterConfig(analyzer);
		iwc.setOpenMode(openMode);
		//IndexWriterConfig indexWriterConfig=new IndexWriterConfig(Version.LUCENE_43,new StandardAnalyzer(Version.LUCENE_43));  
        iwc.setMaxBufferedDocs(1000);  
        iwc.setMergePolicy(mergePolicy);  
        iwc.setRAMBufferSizeMB(64);
        if(indexWriter == null){
           synchronized (lockWirter) {
        	   if(indexWriter == null){
		        	try {
		    			indexWriter = new IndexWriter(this.dir, iwc);
		    		} catch (IOException e) {
		    			throw new DataException(e.getMessage(), e);
		    		}
		        	if(openMode != OpenMode.CREATE){
		        		Thread t  = new Thread(new Runnable(){

							@Override
							public void run() {
								try {
									if(indexWriter.hasDeletions()){
										indexWriter.forceMergeDeletes();
									}
									indexWriter.maybeMerge();
								} catch (IOException e) {
									e.printStackTrace();
								}
							}
		        		});
		        		t.start();
		        	}
        	   }
        	}
		}
		
		return this;
	}
	
	@Override
	public long getToken(IndexSearcher searcher) throws IOException{
	   // Record the current searcher, and save the returend
	   // token into user's search results (eg as a  hidden
	   // HTML form field):
	   long token = this.lifetimeMGR.record(searcher);
	   return token;
	}
	
	
	@Override
	public <T> T search(final SearchHandler<T> handler) throws IOException{
		IndexSearcher searcher = this.acquire();
		T t = null;
		try {
			t = handler.handl(searcher);
		} finally {
			this.release(searcher);
			// Do not use searcher after this!
			searcher = null;
		}
		return t;
	}
	
	
	@Override
	public <T> T search(final long token, final SearchHandler<T> handler) throws IOException{
		// If possible, obtain the same searcher as the last
		// search:
		IndexSearcher searcher = lifetimeMGR.acquire(token);
		if (searcher == null) {
			// Searcher was pruned -- notify user session timed
			// out, or, pull fresh searcher again
			searcher = this.acquire();
		}
		T t = null;
		try {
			t = handler.handl(searcher);
		} finally {
			try {
				lifetimeMGR.release(searcher);
			} catch (IOException e) {
				e.printStackTrace();
			}
			// Do not use searcher after this!
			searcher = null;
		}
		return t;
	}
	
	
	@Override
	public boolean reopen() {
		try {
			this.close();
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return this.open();
	}

	@Override
	public boolean reopen(String address) {
		try {
			this.close();
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return this.open(address);
	}

	@Override
	public boolean reopen(String address, Analyzer analyzer, OpenMode openMode) {
		try {
			this.close();
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return this.open(address, analyzer, openMode);
	}

	@Override
	public Analyzer getDefaultAnalyzer() {
		return AnalyzerUtils.getInstance();
	}

	@Override
	public void addDoc(Document doc) {
		IndexWriter indexWriter = null;
		try {
			indexWriter = this.getIndexWriter();
			indexWriter.addDocument(doc);
			if (this.isAutoCommit()) {
				indexWriter.commit();
			}
		} catch (IOException e) {
			if (this.isAutoCommit()) {
				try {
					indexWriter.rollback();
				} catch (IOException e1) {
					throw new DataException(e1.getMessage(), e1.getCause());
				}
			}
			throw new DataException(e.getMessage(), e);
		} finally {

		}
	} 
	
	
	@Override
	public void addDoc(List<IndexableField> list) {
		IndexWriter indexWriter = null;
		try {
			indexWriter = this.getIndexWriter();
			Document doc = new Document();
			for (IndexableField indexableField : list) {
				doc.add(indexableField);
			}
			this.addDoc(doc);
			if (this.isAutoCommit()) {
				indexWriter.commit();
			}
		} catch (IOException e) {
			if (this.isAutoCommit()) {
				try {
					indexWriter.rollback();
				} catch (IOException e1) {
					throw new DataException(e1.getMessage(), e1.getCause());
				}
			}
			throw new DataException(e.getMessage(), e);
		} finally {

		}
	}


	@Override
	public void addDocs(List<Document> docs) {
		IndexWriter indexWriter = null;
		try {
			indexWriter = this.getIndexWriter();
			indexWriter.addDocuments(docs);
			if (this.isAutoCommit()) {
				indexWriter.commit();
			}
		} catch (IOException e) {
			if (this.isAutoCommit()) {
				try {
					indexWriter.rollback();
				} catch (IOException e1) {
					throw new DataException(e1.getMessage(), e1.getCause());
				}
			}
			throw new DataException(e.getMessage(), e);
		} finally {

		}
	}

	@Override
	public void deleteAllDoc() {
		IndexWriter indexWriter = null;
		try {
			indexWriter = this.getIndexWriter();
			indexWriter.deleteAll();
			if (this.isAutoCommit()) {
				indexWriter.commit();
			}
		} catch (IOException e) {
			if (this.isAutoCommit()) {
				try {
					indexWriter.rollback();
				} catch (IOException e1) {
					throw new DataException(e1.getMessage(), e1.getCause());
				}
			}
			throw new DataException(e.getMessage(), e);
		} finally {

		}
	}


	@Override
	public void deleteDoc(Query... query) {
		IndexWriter indexWriter = null;
		try {
			indexWriter = this.getIndexWriter();
			indexWriter.deleteDocuments(query);
			if (this.isAutoCommit()) {
				indexWriter.commit();
			}
		} catch (IOException e) {
			if (this.isAutoCommit()) {
				try {
					indexWriter.rollback();
				} catch (IOException e1) {
					throw new DataException(e1.getMessage(), e1.getCause());
				}
			}
			throw new DataException(e.getMessage(), e);
		} finally {

		}
	}

	@Override
	public void deleteDoc(Term... term) {
		IndexWriter indexWriter = null;
		try {
			indexWriter = this.getIndexWriter();
			indexWriter.deleteDocuments(term);
			if (this.isAutoCommit()) {
				indexWriter.commit();
			}
		} catch (IOException e) {
			if (this.isAutoCommit()) {
				try {
					indexWriter.rollback();
				} catch (IOException e1) {
					throw new DataException(e1.getMessage(), e1.getCause());
				}
			}
			throw new DataException(e.getMessage(), e);
		} finally {

		}
	}

	@Override
	public TopDocs searchTopDocs(IndexSearcher searcher, final SearchCondition searchCondition) {
		TopScoreDocCollector topCollector = TopScoreDocCollector.create(searchCondition.getMaxDoc());//一个集合
		try {
			searcher.search(searchCondition.getQuery(), topCollector);
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return topCollector.topDocs();
	}
	
	
	@Override
	public TopDocs search(final SearchCondition searchCondition) {
		TopDocs topdocs = null;
		try {
			topdocs = this.search(new SearchHandler<TopDocs>() {
				
				@Override
				public TopDocs handl(IndexSearcher searcher) throws IOException {
					TopDocs topdocs = null;
					if(searchCondition.getLastPageLastDoc()!=null){
						if(searchCondition.getSort() != null){
							topdocs = searcher.searchAfter(searchCondition.getLastPageLastDoc()
									, searchCondition.getQuery()
									, searchCondition.getMaxDoc()
									, searchCondition.getSort(), searchCondition.isDoDocScores()
									, searchCondition.isDoMaxScore());
						}else{
							topdocs = searcher.searchAfter(searchCondition.getLastPageLastDoc()
									, searchCondition.getQuery()
									,searchCondition.getMaxDoc());
						}
					}else{
						if(searchCondition.getSort() != null){
							topdocs = searcher.search(searchCondition.getQuery()
									, searchCondition.getMaxDoc()
									, searchCondition.getSort(), searchCondition.isDoDocScores()
									, searchCondition.isDoMaxScore());
						}else{
							topdocs = searcher.search(searchCondition.getQuery()
									,searchCondition.getMaxDoc());
						}
					}
					return topdocs;
				}
			});
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return topdocs;
	}
	
	@Override
	public <T> List<T> scan(final SearchCondition searchCondition, final DocMapper<T> mapper){
		final List<T> list = new ArrayList<T>();
		this.scan(searchCondition, new TopDocMapper() {
			
			@Override
			public boolean mapper(IndexSearcher searcher,TopDocs doc, int page) throws DataException {
				try {
					List<T> templist = LuceneUtil.getResult(0, searchCondition.getPageSize()
							, searcher, doc,searchCondition.getShowFields(), mapper);
					if(templist != null){
						list.addAll(templist);
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
				return true;
			}
		});
		return list;
	}
		
	@Override
	public long scan(final SearchCondition searchCondition, final TopDocMapper mapper){
		long count = 0L;
		final SearchCondition sc = (SearchCondition) searchCondition.clone();
		try {
			count = this.search(new SearchHandler<Long>() {
				@Override
				public Long handl(IndexSearcher searcher) throws IOException {
					return scan(searcher, sc, mapper);
				}
			});
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return count;
	}
	
	 
	@Override
	public void updateDoc(Term term, Document doc) {
		IndexWriter indexWriter = null;
		try {
			indexWriter = this.getIndexWriter();
			indexWriter.updateDocument(term, doc);
			if (this.isAutoCommit()) {
				indexWriter.commit();
			}
		} catch (IOException e) {
			if (this.isAutoCommit()) {
				try {
					indexWriter.rollback();
				} catch (IOException e1) {
					throw new DataException(e1.getMessage(), e1.getCause());
				}
			}
			throw new DataException(e.getMessage(), e);
		} finally {

		}

	}

	public SearcherManager getSearcherManager() {
		return searcherManager;
	}

	public void setSearcherManager(SearcherManager searcherManager) {
		this.searcherManager = searcherManager;
	}

	@Override
	public boolean isNearRealTime() {
		return nearRealTime;
	}

	public DefaultLuceneServer setNearRealTime(boolean nearRealTime) {
		this.nearRealTime = nearRealTime;
		return this;
	}
	

	@Override
	public void release(IndexSearcher indexSearcher) {
		if (indexSearcher != null) {
			try {
				searcherManager.release(indexSearcher);
			} catch (IOException e) {
				throw new DataException(e.getMessage(), e);
			}
		} 
	}

	@Override
	public IndexSearcher acquire() {
		try {
			return searcherManager.acquire();
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
	}
	

	public long getTime() {
		return time;
	}

	public void setTime(long time) {
		this.time = time;
	}
	@Override
	public String toString() {
		return "LuceneServerImpl [address=" + address + ", opened=" + opened
				+ ", nearRealTime=" + nearRealTime + ", time=" + time + "]";
	}

	@Override
	public void refushTimestamp(){
		time = System.currentTimeMillis();
		try {
//			this.searcherManager.maybeRefreshBlocking();
			this.searcherManager.maybeRefresh();
			this.logger.debug(this.getAddress()+"刷新耗时："+(System.currentTimeMillis()-time));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	 
	@Override
	public long count(final SearchCondition searchCondition){
		Long count = 0L;
		try {
			count = this.search(new SearchHandler<Long>() {

				@Override
				public Long handl(IndexSearcher searcher) throws IOException {
					return count(searcher, searchCondition);
				}
			});
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return count;
	}

	
	public ExecutorService getService() {
		return service;
	}

	public void setService(ExecutorService service) {
		if(service != null){
			service.shutdown();
		}
		this.service = service;
	}
	
}
